package com.leo.hadoop.map;

import com.leo.hadoop.map.*;
import com.leo.util.*;

import java.io.IOException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.conf.Configuration;

/**
 * Describe class ConditionMapper here.
 *
 *
 * Created: Tue Apr 16 22:33:45 2013
 *
 * @author <a href="mailto:leo@leo-pc">Leo</a>
 * @version 1.0
 */
public class ConditionMapper extends MyMapper {

    /**
     * Creates a new <code>ConditionMapper</code> instance.
     *
     */
    public ConditionMapper() {

    }

    @Override
    public void process(Text key, Text value, Context context) throws IOException, InterruptedException {

	String line = value.toString();
	
	String year = line.substring(15, 19);
	int airTemperature;
	if (line.charAt(40) == '+') {// parseInt doesn't like leading plus signs
	    airTemperature = Integer.parseInt(line.substring(41, 45));
	} else {
	    airTemperature = Integer.parseInt(line.substring(40, 45));
	}
	String quality = line.substring(45, 46);

	context.write(new Text(year), new Text(year));

	
    }
	
    @Override
    public Schema getOutputSchema() {
	return this.schema;
    }

    @Override
    protected void init(Configuration conf) {
    }
}
